package com.leo.hbase.mapreduce.service;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Put;

import java.io.IOException;
import java.util.List;

/**
 * @author leo.jie (weixiao.me@aliyun.com)
 * @version 1.0
 * @organization CC
 * @website https://www.jlpyyf.com
 * @date 2019/11/12 11:11 下午
 * @since 1.0
 */
public class HbaseClientService {
    private String rootDir;
    private String zkServer;
    private String port;
    private Configuration conf;
    private Connection hConn = null;
    private Admin admin = null;

    public HbaseClientService(String rootDir, String zkServer, String port) {
        super();
        this.rootDir = rootDir;
        this.zkServer = zkServer;
        this.port = port;
        conf = HBaseConfiguration.create();
        conf.set("hbase.rootdir", rootDir);
        conf.set("hbase.zookeeper.quorum ", zkServer);
        conf.set("hbase.zookeeper.property.clientPort", port);
        try {
            hConn = ConnectionFactory.createConnection(conf);
            admin = hConn.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void saveData(String tableName, List<Put> puts) {
    }

    public void createTable(String tableName, List<String> families) {
        TableName table = TableName.valueOf(tableName);
        try {
            if (admin.tableExists(table)) {
                System.out.println(tableName + " 已经存在！");
            } else {
                HTableDescriptor tableDescriptor = new HTableDescriptor(table);
                for (String family : families) {
                    HColumnDescriptor columnDescriptor = new HColumnDescriptor(family);
                    tableDescriptor.addFamily(columnDescriptor);
                }
                admin.createTable(tableDescriptor);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public <T> void close(T... ts) {
        for (T t : ts) {
            //循环关闭
            try {
                if (t instanceof AutoCloseable) {
                    ((AutoCloseable) t).close();
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
